package com.shujia.streaming

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}


/**
 *  注意事项：
 *    1、spark streaming程序中，每接收到一批数据，都会产生一个job作业
 *    2、spark streaming程序，至少需要2个核线程执行，一个线程进行接收数据，一个线程进行处理数据
 */
object WordCountDemo1 {
  def main(args: Array[String]): Unit = {
    /**
     * Spark core:        SparkContext -> RDD
     * Spark Sql:         SparkSession -> DataFrame
     * Spark Streaming:   StreamingContext -> DStream
     */
    //this(sparkContext: SparkContext, batchDuration: Duration)

    val ss: SparkSession = SparkSession.builder()
      .master("local[*]") // 如果不给线程核数，默认只有1个
      .appName("word count")
      .getOrCreate()

    val sc: SparkContext = ss.sparkContext

    //    val conf = new SparkConf()
    //    conf.setMaster("local")
    //    conf.setAppName("word count")
    //    val sc = new SparkContext(conf)

    //传入一个SparkContext对象，以及接收要处理的数据时间段
    val streamingContext = new StreamingContext(sc, Durations.seconds(5L))

    //近实时处理中，数据应当是源源不断的
    //使用端口号的数据，模拟源源不断的数据源
    //监控对应端口号中的数据
    val infoDS: ReceiverInputDStream[String] = streamingContext.socketTextStream("master", 12345)

    // hello world hello java
    //因为DStream中封装的是RDD,所以可以使用RDD中的大部分的算子
    val resDS: DStream[(String, Int)] = infoDS.flatMap(_.split(" "))
      .map((_, 1))
      .reduceByKey(_ + _)

    println("-----------------------------------")
    resDS.print()
    println("-----------------------------------")

    /**
     * 额外提交命令，一直启动sparkstreaming程序
     */
    streamingContext.start()
    streamingContext.awaitTermination()
    streamingContext.stop()

  }
}
